package com.etlmaster.executor.steps.scheduler;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.List;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;

import com.etlmaster.executor.Step;
import com.etlmaster.executor.StepMeta;
import com.etlmaster.executor.agent.Agents;
import com.etlmaster.executor.agent.ZkAgent;
import com.etlmaster.executor.annotation.StepInstruction;
import com.etlmaster.executor.scheduler.ScheduleServer;
import com.etlmaster.executor.scheduler.utils.LogClusterInfoUtil;
import com.etlmaster.executor.scheduler.utils.ZkUtils;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.StepState;


/**
 * @author Larry
 * @copyright Larry@etlMaster
 * @since 2017-09-10
 */
@StepInstruction("addDefaultZkAgent")

public class AddZkAgentStep extends Step{

	private AddZkAgentStepMeta meta;

	private ZooKeeper zk;
	private Watcher wh;
	public AddZkAgentStep(StepMeta meta) {
		super(meta);
		this.meta=(AddZkAgentStepMeta) meta;
	}

	@Override
	protected int handle() {
		int flag=StepState.RUNNING_SUCCESSED;
		try {
			wh = ZkUtils.getWatcher();
			zk = ZkUtils.getInstance(
					DataCollectionUtils.getProperties("zookeeper.connect"),
					DataCollectionUtils.getIntProperties("zktimeout"), wh);

		
			Stat stat = zk.exists(meta.getZkPath(), false);

			if (stat != null) {
				List<String> children = zk.getChildren(meta.getZkPath(), false);
				if (children.size() > 0) {
					for (String child : children) {
						addZkAgent(child);
					}
				}
			}else{
				flag= StepState.RUNNING_FAILED;
			}

		} catch (IOException e) {
			flag= StepState.RUNNING_FAILED;
		} catch (KeeperException e) {
			flag= StepState.RUNNING_FAILED;

		} catch (InterruptedException e) {
			flag= StepState.RUNNING_FAILED;

		}catch (Exception e) {
			e.printStackTrace();
			flag= StepState.RUNNING_FAILED;

		}
		meta.log("INFO","agent 大小为{}",ScheduleServer.agents.size()+"");

		LogClusterInfoUtil.batchUpdate();
		meta.getStepState().getReturnValue().put("agent_size", ScheduleServer.agents.size());
		return StepState.RUNNING_SUCCESSED;
	}

	private void addZkAgent(String agentId)
			throws UnsupportedEncodingException, KeeperException,
			InterruptedException {

		String host = new String(zk.getData(meta.getZkPath() + "/" + agentId,
				true, null), "utf-8");
		Agents agent = new ZkAgent(agentId,host,
				DataCollectionUtils.getIntProperties("agentMaxThread"));
		LogClusterInfoUtil.insertNewHost(agentId, host, "agent", 0);
		ScheduleServer.agents.put(agentId, agent);

	}
	@Override
	public boolean stop() {
		// TODO Auto-generated method stub
		return false;
	}

}
